[VL] The POC of supporting Flink in Gluten #8839

shuai-xu · 2025-02-27T02:05:41Z

What changes were proposed in this pull request?

This pr is the Java side of POC to support Flink. It generates a GlutenCalOperator to run filter using native.
The draft design is here.

github-actions · 2025-02-27T02:05:59Z

Thanks for opening a pull request!

Could you open an issue for this pull request on Github Issues?

https://github.com/apache/incubator-gluten/issues

Then could you also rename commit message and pull request title in the following format?

[GLUTEN-${ISSUES_ID}][COMPONENT]feat/fix: ${detailed message}

See also:

Other pull requests

PHILO-HE · 2025-02-27T10:09:03Z

@shuai-xu, thanks for your great work! Could you draft a design doc? Google doc is preferred.

zhztheplayer · 2025-02-27T10:21:22Z

gluten-flink/core/src/main/java/org/apache/gluten/backendsapi/FlinkBackend.java

+ */
+package org.apache.gluten.backendsapi;
+
+public class FlinkBackend {


Worth emphasizing that Flink will not be a backend in Gluten. It's more considered a frontend or a framework or so.

OK, We can discuss it.

jinchengchenghh · 2025-02-27T11:20:07Z

Can we use gluten-substrait module? Looks like there exists too much duplicated code.

FelixYBW · 2025-02-27T19:46:27Z

Thank you for the PR! It eventually starts

majetideepak · 2025-02-27T20:44:31Z

Is there a design for this support for people who are not very familiar with Flink?

weiting-chen · 2025-02-28T02:25:49Z

Is there a design for this support for people who are not very familiar with Flink?

This is an initial PR and require more people to join with us for reviewing the design.
For the background and current potential issues, please check the link: https://docs.google.com/document/d/1VNMs1oR0c02kuQLFLQXZeTyk3CoPjrtAfiKGoLGxGzE/edit?usp=sharing

shuai-xu · 2025-02-28T03:22:01Z

Can we use gluten-substrait module? Looks like there exists too much duplicated code.

Yes, we need to discuss whether it need to translate to substrait plan or just call velox jni interface, if do need to translate to substrait plan, It need to reconstruct the gluten-substraint module a little to share it between gluten spark and flink.

shuai-xu · 2025-02-28T03:25:57Z

@PHILO-HE @majetideepak OK, I will write a design soon.

zhztheplayer · 2025-02-28T03:32:55Z

@PHILO-HE @majetideepak OK, I will write a design soon.

Hi @weiting-chen, would you share who is the author of https://docs.google.com/document/d/1VNMs1oR0c02kuQLFLQXZeTyk3CoPjrtAfiKGoLGxGzE/edit?usp=sharing?

I am opening a GitHub side discussion here: #8849

weiting-chen · 2025-02-28T03:38:22Z

@PHILO-HE @majetideepak OK, I will write a design soon.

Hi @weiting-chen, would you share who is the author of https://docs.google.com/document/d/1VNMs1oR0c02kuQLFLQXZeTyk3CoPjrtAfiKGoLGxGzE/edit?usp=sharing?

I am opening a GitHub side discussion here: #8849

Yes, done to add the author.

...flink/runtime/src/main/java/org/apache/gluten/table/runtime/operators/GlutenCalOperator.java

+        }
+    }
+
+    private native long nativeProcessElement(int executor, long data);


zjuwangg · 2025-03-03T08:55:55Z

Very glad to see such exciting progress on Flink support.
Looking forward on how to run the poc code in test...

ParyshevSergey · 2025-03-04T04:09:38Z

System.load("/home/xushuai/gluten/cpp-flink/build/libgflink.so");
How to get this library? Or which steps need to pass for compile it?

zhztheplayer · 2025-03-04T04:31:38Z

...flink/runtime/src/main/java/org/apache/gluten/table/runtime/operators/GlutenCalOperator.java

+                new BoundSplit(
+                        "5",
+                        -1,
+                        new ExternalStreamConnectorSplit("escs1", es.id())));


Would use connector-external-stream as connector ID.

It is currently a fixed value in Velox4j; https://github.com/velox4j/velox4j/blob/434ae37dfc3d5fb79788fe5bce41e41dd17901b5/src/main/cpp/main/velox4j/init/Init.cc#L94-L97

zhztheplayer · 2025-03-04T04:34:13Z

...flink/runtime/src/main/java/org/apache/gluten/table/runtime/operators/GlutenCalOperator.java

+        ExternalStream es = session.externalStreamOps().bind(new DownIterator(inputIterator));
+        List<BoundSplit> splits = List.of(
+                new BoundSplit(
+                        "5",


The planNodeId is the same with leaf scan node's node ID so Velox knows we bind this split to the scan. Perhaps we can pass the scan node ID into GlutenCalOperator somehow?

zhztheplayer · 2025-03-04T04:41:25Z

gluten-flink/planner/src/main/java/org/apache/gluten/rexnode/RexNodeConverter.java

+            return new ConstantTypedExpr(
+                    toType(literal.getType()),
+                    toVariant(literal),
+                    null);


Could use ConstantTypedExpr.create(toVariant(literal))

The create will call native method, I think we'd better not call native in client side, so we need not load native libraries in Flink client side.

zhztheplayer · 2025-03-04T04:44:06Z

...nner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecCalc.java

+        // add a mock input as velox not allow the source is empty.
+        PlanNode mockInput = new ValuesNode(
+                String.valueOf(ExecNodeContext.newNodeId()),
+                "",
+                false,
+                1);


Adding a scan node so we can bind the split to it. E.g.,

final TableScanNode scanNode = new TableScanNode( "id-1", ...(type), new ExternalStreamTableHandle("connector-external-stream"), List.of() );

zhztheplayer · 2025-03-04T04:45:36Z

gluten-flink/planner/src/main/java/org/apache/gluten/rexnode/RexNodeConverter.java

+            case VARCHAR:
+                return new VarCharValue(literal.getValue().toString());
+            case BINARY:
+                return new VarBinaryValue(literal.getValue().toString());


Should we use VarBinaryValue.create() to pass a byte array in?

ParyshevSergey · 2025-03-05T10:04:55Z

After try to run this POC, got error
Caused by: io.github.zhztheplayer.velox4j.exception.VeloxException: Exception: VeloxUserError Error Source: USER Error Code: INVALID_ARGUMENT Reason: Splits can be associated only with leaf plan nodes which require splits. Plan node ID 5 doesn't refer to such plan node.

shuai-xu · 2025-03-06T03:16:01Z

After try to run this POC, got error Caused by: io.github.zhztheplayer.velox4j.exception.VeloxException: Exception: VeloxUserError Error Source: USER Error Code: INVALID_ARGUMENT Reason: Splits can be associated only with leaf plan nodes which require splits. Plan node ID 5 doesn't refer to such plan node.

You can try the latest code.

ParyshevSergey · 2025-03-06T07:36:54Z

After try to run this POC, got error Caused by: io.github.zhztheplayer.velox4j.exception.VeloxException: Exception: VeloxUserError Error Source: USER Error Code: INVALID_ARGUMENT Reason: Splits can be associated only with leaf plan nodes which require splits. Plan node ID 5 doesn't refer to such plan node.

You can try the latest code.

Thanks for reply. I rerun with last commit and gotcha

Caused by: io.github.zhztheplayer.velox4j.exception.VeloxException: com.fasterxml.jackson.databind.exc.UnrecognizedPropertyException: Unrecognized field "sources" (class io.github.zhztheplayer.velox4j.plan.TableScanNode), not marked as ignorable (4 known properties: "id", "tableHandle", "outputType", "assignments"])
 at [Source: UNKNOWN; byte offset: #UNKNOWN] (through reference chain: io.github.zhztheplayer.velox4j.plan.TableScanNode["sources"])

I looked at TableScanNode and notice that json getter exist on "sources" but this field is absence. I try rebuild velox4j with Serde change (add disable for jackson DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES) and got

Caused by: io.github.zhztheplayer.velox4j.exception.VeloxException: couldn't find key name in dynamic object
	at io.github.zhztheplayer.velox4j.jni.JniWrapper.executeQuery(Native Method)
	at io.github.zhztheplayer.velox4j.jni.JniApi.executeQuery(JniApi.java:52)
	at io.github.zhztheplayer.velox4j.query.Queries.execute(Queries.java:14)
	at org.apache.gluten.table.runtime.operators.GlutenCalOperator.processElement(GlutenCalOperator.java:97)

SQL query

CREATE TABLE srcTbl (id INT, price INT, name STRING) WITH ('connector'='datagen');
CREATE TABLE snkTbl (id INT, price INT) WITH ('connector'='blackhole');
INSERT INTO snkTbl SELECT id, price FROM srcTbl WHERE price > 10;

The POC of supporting Flink in Gluten

396abba

weiting-chen added the enhancement New feature or request label Feb 27, 2025

weiting-chen changed the title ~~The POC of supporting Flink in Gluten~~ [VL] The POC of supporting Flink in Gluten Feb 27, 2025

zhztheplayer reviewed Feb 27, 2025

View reviewed changes

refine the directories to follow flink stytle

e516d49

zjuwangg reviewed Mar 3, 2025

View reviewed changes

...flink/runtime/src/main/java/org/apache/gluten/table/runtime/operators/GlutenCalOperator.java Outdated

}

}

private native long nativeProcessElement(int executor, long data);

This comment was marked as resolved.

Sign in to view

FelixYBW mentioned this pull request Mar 4, 2025

Add initial flink support #8888

Open

shuai-xu added 2 commits March 4, 2025 12:23

use velox4j to execute

3afead5

add mising files

974b972

zhztheplayer reviewed Mar 4, 2025

View reviewed changes

zhztheplayer mentioned this pull request Mar 4, 2025

Make some essential constructors public velox4j/velox4j#80

Merged

Example run pass and add docs

4a15f3f

github-actions bot added the DOCS label Mar 6, 2025

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[VL] The POC of supporting Flink in Gluten #8839

[VL] The POC of supporting Flink in Gluten #8839

shuai-xu commented Feb 27, 2025 •

edited

Loading

github-actions bot commented Feb 27, 2025

PHILO-HE commented Feb 27, 2025

zhztheplayer Feb 27, 2025

shuai-xu Feb 28, 2025

jinchengchenghh commented Feb 27, 2025

FelixYBW commented Feb 27, 2025

majetideepak commented Feb 27, 2025 •

edited

Loading

weiting-chen commented Feb 28, 2025

shuai-xu commented Feb 28, 2025

shuai-xu commented Feb 28, 2025

zhztheplayer commented Feb 28, 2025

weiting-chen commented Feb 28, 2025

This comment was marked as resolved.

zjuwangg commented Mar 3, 2025

ParyshevSergey commented Mar 4, 2025

zhztheplayer Mar 4, 2025 •

edited

Loading

zhztheplayer Mar 4, 2025

zhztheplayer Mar 4, 2025

shuai-xu Mar 4, 2025

zhztheplayer Mar 4, 2025

zhztheplayer Mar 4, 2025

shuai-xu Mar 4, 2025

ParyshevSergey commented Mar 5, 2025

shuai-xu commented Mar 6, 2025

ParyshevSergey commented Mar 6, 2025 •

edited

Loading

[VL] The POC of supporting Flink in Gluten #8839

Are you sure you want to change the base?

[VL] The POC of supporting Flink in Gluten #8839

Conversation

shuai-xu commented Feb 27, 2025 • edited Loading

What changes were proposed in this pull request?

github-actions bot commented Feb 27, 2025

PHILO-HE commented Feb 27, 2025

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jinchengchenghh commented Feb 27, 2025

FelixYBW commented Feb 27, 2025

majetideepak commented Feb 27, 2025 • edited Loading

weiting-chen commented Feb 28, 2025

shuai-xu commented Feb 28, 2025

shuai-xu commented Feb 28, 2025

zhztheplayer commented Feb 28, 2025

weiting-chen commented Feb 28, 2025

This comment was marked as resolved.

zjuwangg commented Mar 3, 2025

ParyshevSergey commented Mar 4, 2025

zhztheplayer Mar 4, 2025 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ParyshevSergey commented Mar 5, 2025

shuai-xu commented Mar 6, 2025

ParyshevSergey commented Mar 6, 2025 • edited Loading

shuai-xu commented Feb 27, 2025 •

edited

Loading

majetideepak commented Feb 27, 2025 •

edited

Loading

zhztheplayer Mar 4, 2025 •

edited

Loading

ParyshevSergey commented Mar 6, 2025 •

edited

Loading